package org.nbict.iot.task.transit;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.StringUtils;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.nbict.iot.protocol.config.ProtocolMapper;
import org.nbict.iot.protocol.context.FieldTypeContext;
import org.nbict.iot.protocol.context.ParserContext;
import org.nbict.iot.protocol.type.ProtocolField;
import org.nbict.iot.protocol.util.BytesOperator;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;

/**
 *
 * 报文解析任务
 * Created by songseven on 18/6/19.
 */
@Deprecated
public class BinaryParserBolt extends BaseBasicBolt {

    private ProtocolField rootField;
    private final static byte[] tag = {0x01};
    private final static byte[] patch = {0x23};

    public void prepare(Map stormConf, TopologyContext context) {
        String resource = "mybatis-config.xml";
        SqlSession session = null;
        try {
            InputStream inputStream = Resources.getResourceAsStream(resource);
            SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
            session = sqlSessionFactory.openSession();
            ProtocolMapper mapper = session.getMapper(ProtocolMapper.class);
            rootField = mapper.getProtocolFieldById(120);
        } catch (IOException e) {
            //log
        } finally {
            session.close();
        }
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

//        String topic = StringUtils.substringAfter((String)input.getValue(0), "-"); //从raw-5vvokv-simple截取5vvokv-simple
        String topic = StringUtils.replaceOnce((String)input.getValue(0), "raw", "transit");
//        Long offset = (Long) input.getValue(2); //offset

        byte[] raw = (byte[]) input.getValue(4);
        byte[] checksum = BytesOperator.splitBytes(raw, 0, 1);

        switch (checksum[0]) {
            case (byte)0x01:
                raw = BytesOperator.splitFrontBytes(raw, 1);
            case (byte)0x23: //打补丁
                try {
                    ParserContext context = new FieldTypeContext(raw, rootField);
                    JSON re = (JSON) context.type().process(context);
                    //System.out.println(re.toJSONString());
                    collector.emit("correct", new Values(topic, re.toJSONString()));
                } catch (Exception e) {
                    //TODO 错误流
                }
                break;
            default:
                //TODO 错误流
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("correct", new Fields("platform_id", "parsed")); //correct
    }
}
